package com.yc.geo.geohash.udf;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.hive.udf.util.JsonUtils;
import com.hive.udf.util.VincentyGeodesy;
import com.yc.geo.geohash.algorithm.GeoHash_V2;
import com.yc.geo.geohash.model.Region;
import com.yc.geo.geohash.model.WGS84Point;
import org.apache.commons.io.IOUtils;
import org.apache.commons.io.LineIterator;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hive.ql.exec.UDF;
import org.codehaus.jackson.JsonNode;
import org.codehaus.jackson.JsonParser;
import org.codehaus.jackson.map.DeserializationConfig;
import org.codehaus.jackson.map.ObjectMapper;
import org.codehaus.jackson.map.SerializationConfig;
import org.codehaus.jackson.map.annotate.JsonSerialize;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.PrintWriter;
import java.text.SimpleDateFormat;
import java.util.List;
import java.util.Map;

/**
 * 转换经纬度为行政区域 : eg: 31.194704,120.715525 =>  {"province":"江苏省", "city":"苏州市","district":"吴江区"}
 * 注： 按距离每个区的距离算，离哪个区中心最近返回哪个区
 * 如果需要准确返回"归属行政区域"，需要调用高德API，效率较低
 * @author zhaoxiang
 */
public class ReGeo extends UDF{
	private static Logger logger = LoggerFactory.getLogger(ReGeo.class);
	private static ObjectMapper objectMapper = new ObjectMapper();
	private static final List<Region> REGIONS = Lists.newLinkedList();
	static {
		objectMapper.setSerializationInclusion(JsonSerialize.Inclusion.NON_NULL); 
		objectMapper.configure(SerializationConfig.Feature.WRITE_DATES_AS_TIMESTAMPS, false);
		objectMapper.getDeserializationConfig().withDateFormat(new SimpleDateFormat("yyyy-MM-dd"));
		objectMapper.getSerializationConfig().withDateFormat(new SimpleDateFormat("yyyy-MM-dd"));
		objectMapper.configure(DeserializationConfig.Feature.FAIL_ON_UNKNOWN_PROPERTIES, false);
		objectMapper.configure(JsonParser.Feature.ALLOW_UNQUOTED_FIELD_NAMES, true);
		objectMapper.configure(JsonParser.Feature.ALLOW_SINGLE_QUOTES, true);
		objectMapper.configure(JsonParser.Feature.ALLOW_UNQUOTED_CONTROL_CHARS, true);
		objectMapper.configure(SerializationConfig.Feature.FAIL_ON_EMPTY_BEANS, false);
		init();
	}
	
	public static void init() {
		//加载行政区域数据
		
		LineIterator lineIterator;
		try {
			lineIterator = IOUtils.lineIterator(ReGeo.class.getClassLoader().getResourceAsStream("region-cn.txt"), "UTF-8");
			while(lineIterator.hasNext()){
				String l = lineIterator.nextLine();
				Region region = JsonUtils.fromJSON(l, Region.class);
				REGIONS.add(region);
			}
			lineIterator.close();
		} catch (IOException e) {
			logger.error("load region-cn.txt error:", e);
			System.exit(127);
		}	
	}
	
	private static final String GEO_PATTERN = "{\"province\":\"%s\", \"city\":\"%s\",\"district\":\"%s\"}";
	/**
	 * @param latitude 纬度
	 * @param longitude 经度
	 * @return 省，市，区，  eg： 广东省,梅州市,五华县
	 */
	public String evaluate(final Double latitude, final Double longitude) {  
        try {
        	Region curRegion = null;
			WGS84Point curPoint = new WGS84Point(latitude, longitude);
			double minDistance = Double.MAX_VALUE;
			for(Region region : REGIONS){
				double lat1 = region.getLat();
				double lon1 = region.getLon();
				//经纬度都相差不多，才计算距离
				if(Math.abs(lat1 - latitude) < 0.4 && Math.abs(lon1 - longitude) < 0.4){
					WGS84Point otherPoint = new WGS84Point(lat1, lon1);
					double distance = VincentyGeodesy.distanceInMeters(curPoint, otherPoint);
//					double distance = MapUtils.distanceInMeters(curPoint, otherPoint);
					if(distance < minDistance){
						minDistance = distance;
						curRegion = region;
					}
				}
			}
			if (curRegion == null) {
				return String.format(GEO_PATTERN, "-","-","-");
			}
			
			String province = curRegion.getProvince();
			String city = curRegion.getCity();
			String district = curRegion.getDistrict();
        	return String.format(GEO_PATTERN, province,city,district);
		} catch (Exception e) {
			//e.printStackTrace();
			//System.err.println(String.format("get district error, lat: %s, lon: %s", latitude, longitude));
			logger.error("get district error, lat: {}, lon:{}", latitude, longitude);
			logger.error("",e);
			return String.format(GEO_PATTERN, "-","-","-");
		} 
    }  
	
	public static void main(String[] args) throws IOException {
		ReGeo reGeo = new ReGeo();
		System.out.println(reGeo.evaluate(22.303095 ,114.16728));
		
		//加载天气数据
		Map<String,String> weatherMap = Maps.newHashMap();
		LineIterator lineIterator = IOUtils.lineIterator(ReGeo.class.getClassLoader().getResourceAsStream("weather.txt"), "UTF-8");	
		while(lineIterator.hasNext()){
			String jsonStr = lineIterator.nextLine();
			JsonNode jsonNode =  objectMapper.readTree(jsonStr);
			String province_district = jsonNode.get("city").asText() + jsonNode.get("area").asText();
			weatherMap.put(province_district,jsonStr);
		}
		lineIterator.close();
		
		PrintWriter writer = new PrintWriter("src/main/resources/result.txt");
		//读订单数据
		lineIterator = IOUtils.lineIterator(ReGeo.class.getClassLoader().getResourceAsStream("order.txt"), "UTF-8");
		long start = System.currentTimeMillis();
		int i=0;
		while(lineIterator.hasNext()){
			String orderLine = lineIterator.nextLine();
			String[] fields = orderLine.split("\t");
			String service_order_id = fields[0];
			String city_en = fields[18];
			double start_latitude = Double.parseDouble(fields[49]);
			double start_longitude = Double.parseDouble(fields[50]);
			Region curRegion = null;
			
			WGS84Point curPoint = new WGS84Point(start_latitude, start_longitude);
			double minDistance = Double.MAX_VALUE;
			for(Region region : REGIONS){
				double lat1 = region.getLat();
				double lon1 = region.getLon();
				//经纬度都相差不多，才计算距离
				if(Math.abs(lat1 - start_latitude) < 0.4 && Math.abs(lon1 - start_longitude) < 0.4){
					WGS84Point otherPoint = new WGS84Point(lat1, lon1);
					double distance = VincentyGeodesy.distanceInMeters(curPoint, otherPoint);
					if(distance < minDistance){
						minDistance = distance;
						curRegion = region;
					}
				}
			}
			if (curRegion == null) {
				System.err.println(String.format("%s,%s,%s,%s",GeoHash_V2.encode(start_latitude, start_longitude, 7).toHashString(),city_en, start_latitude,start_longitude));
				i++;
				continue;
			}
			
			String province = curRegion.getProvince();
			String city = curRegion.getCity();
			String district = curRegion.getDistrict();
			String weather = weatherMap.get(province + district);
			if(weather == null){
				weather = weatherMap.get(province + city + district);
				if(weather == null){
					weather = weatherMap.get(province + city);
				}
			}
			String[] rs =new String[]{service_order_id, city_en,start_latitude +"",start_longitude+"",province,city, district,weather};
			writer.println(StringUtils.join(rs, ","));
//				String gaodeRs = reGeo.evaluate(start_longitude,start_latitude);
//				if(! gaodeRs.equals(String.format("%s,%s,%s", province,city,district)) ){
//					error_writer.println(String.format("%s,%s == 高德 ：%s  ==  距离计算：%s",start_latitude, start_longitude, gaodeRs, String.format("%s,%s,%s", province,city,district)));
//				}
//			String district =reGeo.evaluate(start_longitude,start_latitude);
//			String[] province_city_district = district.split(",",-1);
//			String weather = weatherMap.get(province_city_district[0]+province_city_district[2]);
//			if(weather == null){
//				weather = weatherMap.get(province_city_district[0]+province_city_district[1]+province_city_district[2]);
//				if(weather == null){
//					weather = weatherMap.get(province_city_district[0]+province_city_district[1]);
//				}
//			}
//			String[] rs =new String[]{service_order_id, city,start_latitude +"",start_longitude+"",district,weather};
//			writer.println(StringUtils.join(rs, " , "));
		}
		writer.close();
		long end = System.currentTimeMillis();
		System.out.println("totail time " + (end - start) +"ms !");
		System.out.println("totail not fount " + i);
	}
}
